package com.hudson.store.consumer;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.reflect.TypeToken;
import com.hudson.core.json.AtomicIntegerTypeAdapter;
import com.hudson.store.concurrent.ThreadPoolConstant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Type;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/7/13 1:21
 **/
public class ConsumerOffset {
    private static final Logger logger = LoggerFactory.getLogger(ConsumerOffset.class);
    // 在你的类中定义 gson 实例
    private static final Gson gson = new GsonBuilder().registerTypeAdapter(AtomicInteger.class, new AtomicIntegerTypeAdapter()).create();
    public static ConcurrentHashMap<String, ConcurrentHashMap<String, Map<Integer, AtomicInteger>>> consumer_offset = null;
    private static String dir = "store/consumer";
    private static String fileName = "consumerOffset.json";
    private static File jsonFile = null;

    static {

        // 创建 store 子目录（如果不存在）
        File storeDir = new File(dir);
        if (!storeDir.exists()) {
            boolean mkdirsSuccess = storeDir.mkdirs();
            if (!mkdirsSuccess) {
                throw new RuntimeException("无法创建目录: " + storeDir.getAbsolutePath());
            }
        }

        // 定义目标文件
        jsonFile = new File(storeDir, fileName);

        // 如果文件不存在，则创建它
        if (!jsonFile.exists()) {
            try {
                boolean createSuccess = jsonFile.createNewFile();
                if (!createSuccess) {
                    throw new RuntimeException("无法创建文件: " + jsonFile.getAbsolutePath());
                }
            } catch (IOException e) {
                throw new RuntimeException("创建 consumerOffset.json 文件失败", e);
            }
        }

        if (jsonFile.length() > 0) {
            try (RandomAccessFile raf = new RandomAccessFile(jsonFile, "r")) {
                long length = raf.length();
                if (length != 0) {
                    byte[] bytes = new byte[(int) length];
                    raf.readFully(bytes);
                    String json = new String(bytes, StandardCharsets.UTF_8);
                    // 使用 TypeToken 解析嵌套泛型
                    Type type = new TypeToken<ConcurrentHashMap<String, ConcurrentHashMap<String, Map<Integer, AtomicInteger>>>>() {
                    }.getType();

                    consumer_offset = gson.fromJson(json, type);
                } else {
                    consumer_offset = new ConcurrentHashMap<>();
                }

            } catch (IOException e) {
                throw new RuntimeException("读取文件失败: " + jsonFile.getAbsolutePath(), e);
            }
        } else {
            consumer_offset = new ConcurrentHashMap<>();
        }

        ThreadPoolConstant.scheduledThreadPoolExecutor.scheduleWithFixedDelay(()->{
            try {
                // 每次都重新打开文件，避免文件锁定问题
                RandomAccessFile randomAccessFile = new RandomAccessFile(jsonFile, "rw");
                FileChannel channel = randomAccessFile.getChannel();

                // 获取 JSON 字符串
                String json = gson.toJson(consumer_offset);
                byte[] bytes = json.getBytes(StandardCharsets.UTF_8);

                // 创建 buffer 并写入
                ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
                buffer.put(bytes);
                buffer.flip(); // 必须 flip，切换为读模式

                // 清空文件内容
                channel.truncate(0); // 删除旧内容
                channel.position(0); // 将写入位置移到开头

                // 写入新内容
                channel.write(buffer);

                // 强制刷盘（可选）
                channel.force(true);

                // 关闭资源
                channel.close();
                randomAccessFile.close();

                logger.info("offset 文件已更新");

                TimeUnit.SECONDS.sleep(5);
            } catch (Exception e) {
                if (e instanceof FileNotFoundException) {
                    try {
                        jsonFile.createNewFile();
                    } catch (IOException ex) {
                        throw new RuntimeException(ex);
                    }
                }
                logger.error("写入 offset 文件失败", e);
            }

        },0,5,TimeUnit.SECONDS);
    }

    public static int getOffset(String consumerGroup, String topic, int queueId) {
        ConcurrentHashMap<String, Map<Integer, AtomicInteger>> consumerGroupMap = consumer_offset.computeIfAbsent(consumerGroup, k -> new ConcurrentHashMap<>());

        Map<Integer, AtomicInteger> atomicIntegerMap = consumerGroupMap.computeIfAbsent(topic, k -> new HashMap<>());

        AtomicInteger atomicInteger = atomicIntegerMap.computeIfAbsent(queueId, k -> new AtomicInteger(0));
        return atomicInteger.get();
    }

    public static void putOffset(String consumerGroup, String topic, int queueId) {
        ConcurrentHashMap<String, Map<Integer, AtomicInteger>> consumerGroupMap = consumer_offset.computeIfAbsent(consumerGroup, k -> new ConcurrentHashMap<>());
        Map<Integer, AtomicInteger> atomicIntegerMap = consumerGroupMap.computeIfAbsent(topic, k -> new ConcurrentHashMap<>());
        atomicIntegerMap.computeIfAbsent(queueId, k -> new AtomicInteger(0)).getAndIncrement();
    }

    public static void putOffset(String consumerGroup, String topic, int queueId, int consumerOffset) {
        ConcurrentHashMap<String, Map<Integer, AtomicInteger>> consumerGroupMap = consumer_offset.computeIfAbsent(consumerGroup, k -> new ConcurrentHashMap<>());
        Map<Integer, AtomicInteger> atomicIntegerMap = consumerGroupMap.computeIfAbsent(topic, k -> new ConcurrentHashMap<>());
        atomicIntegerMap.put(queueId, new AtomicInteger(consumerOffset));
    }
}
